package com.kr.weitao.util;

import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/** Created by maoweidong on 2017/10/17. */
public class KafkaUtil {
  public static final String SUCCESS_CODE = "00000";
  public static final String SUCCESS_MES = "成功";
  /*kakfa-code*/
  public static final String KAFKA_SEND_ERROR_CODE = "30001";
  public static final String KAFKA_NO_RESULT_CODE = "30002";
  public static final String KAFKA_NO_OFFSET_CODE = "30003";
  public static final String KAFKA_SEND_ERROR_MES = "发送消息超时,联系相关技术人员";
  public static final String KAFKA_NO_RESULT_MES = "未查询到返回结果,联系相关技术人员";
  public static final String KAFKA_NO_OFFSET_MES = "未查到返回数据的offset,联系相关技术人员";

  public static Map<String, Object> checkKafkaRecord(
          ListenableFuture<SendResult<String, String>> res) {
    Map<String, Object> m = new HashMap<String, Object>();
    if (res != null) {
      try {
        SendResult r = res.get(); //检查result结果集
        /*检查recordMetadata的offset数据，不检查producerRecord*/
        Long offsetIndex = r.getRecordMetadata().offset();
        if (offsetIndex != null && offsetIndex >= 0) {
          m.put("code", SUCCESS_CODE);
          m.put("message", SUCCESS_MES);
          return m;
        } else {
          m.put("code", KAFKA_NO_OFFSET_CODE);
          m.put("message", KAFKA_NO_OFFSET_MES);
          return m;
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
        m.put("code", KAFKA_SEND_ERROR_CODE);
        m.put("message", KAFKA_SEND_ERROR_MES);
        return m;
      } catch (ExecutionException e) {
        e.printStackTrace();
        m.put("code", KAFKA_SEND_ERROR_CODE);
        m.put("message", KAFKA_SEND_ERROR_MES);
        return m;
      }
    } else {
      m.put("code", KAFKA_NO_RESULT_CODE);
      m.put("message", KAFKA_NO_RESULT_MES);
      return m;
    }
  }
}
